package com.sumbo.config.writer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.Future;

/**
 * \* User: MeiZhongHao
 * \* Date: 2019-06-10
 * \* Time: 10:26
 * \* Description: kafka输出
 * \
 */
public class KafkaWriter extends Writer {

    private KafkaWriter() {
    }

    public static KafkaWriter instance = new KafkaWriter();

    private Map<String, KafkaProducer> producerMap = new HashMap<String, KafkaProducer>();

    public void init(Map<String, Properties> temp) {
        if (temp == null || temp.size() <= 0) {
            return;
        }
        for (String key : temp.keySet()) {
            KafkaProducer consumer = new KafkaProducer<String, String>(temp.get(key));
            producerMap.put(key, consumer);
        }
    }

    public KafkaProducer<String, String> getProducer(String client) {
        return producerMap.get(client);
    }

    public boolean containsProducer(String client) {
        return producerMap.containsKey(client);
    }

    public boolean send(KafkaOutputContext context, String message) {
        String topicsStr = context.getTopic();
        String[] topics = null;
        if (topicsStr == null || (topics = topicsStr.split(",")) == null || topics.length <= 0) {
            return true;
        }
        for (String topic : topics) {
            try {
                send(context.getClient(), topic, message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return true;
    }

    public RecordMetadata send(String client, String topic, String message) throws Exception {
        KafkaProducer producer = getProducer(client);
        if (producer == null) {
            throw new Exception(" No producer named " + client);
        }
        Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>(topic,
                UUID.randomUUID().toString(),
                message));
        return future.get();
    }

    public Map<String, KafkaProducer> getProducerMap() {
        return producerMap;
    }
}